package com.suxin.splitter.executor;

import com.google.common.collect.Lists;
import com.suxin.splitter.MethodExecutor;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author Tang
 * @classname ParallelMethodExecutor
 * @description [ 并行执行器 ]
 * @date 2022/11/7 13:54
 */
public class ParallelMethodExecutor extends AbstractMethodExecutor implements MethodExecutor {

    private final ExecutorService executor;

    private final int taskPreThread;

    public ParallelMethodExecutor(ExecutorService executor, int taskPreThread) {
        this.executor = executor;
        this.taskPreThread = taskPreThread;
    }

    @Override
    protected <P,R> List<R> doExecute(Function<P, R> function, List<P> ps) throws Exception {
        List<Task> tasks = Lists.partition(ps, this.taskPreThread).stream()
                .map(p -> new Task(function, p))
                .collect(Collectors.toList());
        if (tasks.size() == 1) {
            return tasks.get(0).call();
        }

        List<Future<List<R>>> futures = Lists.newArrayListWithCapacity(tasks.size());
        for (int i = 0; i < tasks.size(); i++) {
            futures.add(this.executor.submit(tasks.get(i)));
        }

        List<List<R>> batchResult = Lists.newArrayListWithCapacity(8);
        // 获取所有的结果
        batchResult.addAll(getResultFromFuture(futures));

        List<R> result = Lists.newArrayListWithCapacity(batchResult.size());
        for (List<R> rs : batchResult) {
            result.addAll(rs);
        }
        return result;
    }

    private <R> Collection<? extends List<R>> getResultFromFuture(List<Future<List<R>>> futures) {
        return futures.stream().map(future -> {
            try {
                return future.get();
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }).collect(Collectors.toList());
    }

    static class Task<P, R> implements Callable<List<R>> {
        private final Function<P, R> function;
        private final List<P> params;

        public Task(Function<P, R> function, List<P> ps) {
            this.function = function;
            this.params = ps;
        }

        @Override
        public List<R> call() throws Exception {
            return params.stream()
                    .map(p -> function.apply(p))
                    .collect(Collectors.toList());
        }
    }

}
